package com.roy.minimq.namesrv.infrastructure;

import com.roy.minimq.common.broker.RegisterBrokerParam;
import com.roy.minimq.namesrv.domain.entity.BrokerData;
import com.roy.minimq.namesrv.domain.entity.BrokerLiveInfo;
import com.roy.minimq.namesrv.domain.entity.ClusterInfo;
import com.roy.minimq.namesrv.domain.entity.RegisterBrokerResult;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * TODO 路由关系管理
 * 2021年8月20日 暂未设计清楚要如何维护broker路由信息。
 */
public class RouteInfoManager {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;

    public RouteInfoManager() {
        brokerAddrTable = new HashMap<>(1024);
        clusterAddrTable = new HashMap<>(1024);
        brokerLiveTable = new HashMap<>(1024);
    }

    public byte[] getAllClusterInfo() {
        ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
        clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
        clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
        return clusterInfoSerializeWrapper.encode();
    }

    public void onChannelDestroy(String remoteAddr, Channel channel) {
        logger.info("RouteInfoManager.onChannelDestroy( {},{})", remoteAddr, channel);
        String brokerAddrFound = null;
        if (channel != null) {
            try {
                try {
                    this.lock.readLock().lockInterruptibly();
                    Iterator<Map.Entry<String, BrokerLiveInfo>> itBrokerLiveTable =
                            this.brokerLiveTable.entrySet().iterator();
                    while (itBrokerLiveTable.hasNext()) {
                        Map.Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();
                        if (entry.getValue().getChannel() == channel) {
                            brokerAddrFound = entry.getKey();
                            break;
                        }
                    }
                } finally {
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) {
                logger.error("onChannelDestroy Exception", e);
            }
        }

        if (null == brokerAddrFound) {
            brokerAddrFound = remoteAddr;
        } else {
            logger.info("the broker's channel destroyed, {}, clean it's data structure at once", brokerAddrFound);
        }

        if (brokerAddrFound != null && brokerAddrFound.length() > 0) {

            try {
                try {
                    this.lock.writeLock().lockInterruptibly();
                    this.brokerLiveTable.remove(brokerAddrFound);
                    String brokerNameFound = null;
                    boolean removeBrokerName = false;
                    Iterator<Map.Entry<String, BrokerData>> itBrokerAddrTable =
                            this.brokerAddrTable.entrySet().iterator();
                    while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
                        BrokerData brokerData = itBrokerAddrTable.next().getValue();

                        Iterator<Map.Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();
                        while (it.hasNext()) {
                            Map.Entry<Long, String> entry = it.next();
                            Long brokerId = entry.getKey();
                            String brokerAddr = entry.getValue();
                            if (brokerAddr.equals(brokerAddrFound)) {
                                brokerNameFound = brokerData.getBrokerName();
                                it.remove();
                                logger.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                                        brokerId, brokerAddr);
                                break;
                            }
                        }

                        if (brokerData.getBrokerAddrs().isEmpty()) {
                            removeBrokerName = true;
                            itBrokerAddrTable.remove();
                            logger.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
                                    brokerData.getBrokerName());
                        }
                    }

                    if (brokerNameFound != null && removeBrokerName) {
                        Iterator<Map.Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
                        while (it.hasNext()) {
                            Map.Entry<String, Set<String>> entry = it.next();
                            String clusterName = entry.getKey();
                            Set<String> brokerNames = entry.getValue();
                            boolean removed = brokerNames.remove(brokerNameFound);
                            if (removed) {
                                logger.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                                        brokerNameFound, clusterName);

                                if (brokerNames.isEmpty()) {
                                    logger.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                                            clusterName);
                                    it.remove();
                                }

                                break;
                            }
                        }
                    }
                } finally {
                    this.lock.writeLock().unlock();
                }
            } catch (Exception e) {
                logger.error("onChannelDestroy Exception", e);
            }
        }
    }

    //TODO 注册Broker。这里只实现简单的broker信息注册。
    // 2021.08.23暂时不知道要怎么去简化业务方法。这里就先留着把。
    //clusterName，brokerName,brokerId来源于Broker端的本地配置。 brokerAddr，来自于Borker端解析出来的本地IP。
    public RegisterBrokerResult registerBroker(RegisterBrokerParam registerBrokerParam,
                                               Channel channel) {
        logger.info("RouteinfoManager, broker registered");
        final String clusterName = registerBrokerParam.getClusterName();
        final String brokerName = registerBrokerParam.getBrokerName();
        final Long brokerId = registerBrokerParam.getBrokerId();
        final String brokerAddr = registerBrokerParam.getBrokerAddr();
        try {
            this.lock.writeLock().lockInterruptibly();
            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
            if (null == brokerNames) {
                brokerNames = new HashSet<String>();
            }
            brokerNames.add(brokerName);
            this.clusterAddrTable.put(clusterName, brokerNames);

            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            if (null == brokerData) {
                brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
                this.brokerAddrTable.put(brokerName, brokerData);
            }
            HashMap<Long, String> brokerAddrs = brokerData.getBrokerAddrs();
            brokerAddrs.put(brokerId, brokerAddr);

            final BrokerLiveInfo brokerLiveInfo = new BrokerLiveInfo(System.currentTimeMillis(), channel);
            this.brokerLiveTable.put(brokerAddr, brokerLiveInfo);

        } catch (Exception e) {
            logger.error("registerBroker Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }

        return null;
    }

    public void unregisterBroker(RegisterBrokerParam registerBrokerParam) {
        logger.info("RouteinfoManager, broker unregistered");
        final String clusterName = registerBrokerParam.getClusterName();
        final String brokerName = registerBrokerParam.getBrokerName();
        final Long brokerId = registerBrokerParam.getBrokerId();
        final String brokerAddr = registerBrokerParam.getBrokerAddr();
        try {
            this.lock.writeLock().lockInterruptibly();
            final BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddr);
            logger.info("unregisterBroker, remove from brokerLiveTable {},{}"
                    , null == brokerLiveInfo ? "SUCCED" : "FAILED"
                    , brokerLiveInfo);

            boolean removeBrokerName = false;
            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            if (null != brokerData) {
                final String addr = brokerData.getBrokerAddrs().remove(brokerId);
                logger.info("unregisterBroker, remove addr from brokerAddrTable {}, {}",
                        addr != null ? "OK" : "Failed",
                        brokerAddr
                );

                if(brokerData.getBrokerAddrs().isEmpty()){
                    this.brokerAddrTable.remove(brokerName);
                    logger.info("unregisterBroker, remove name from brokerAddrTable OK, {}",
                            brokerName
                    );
                    removeBrokerName = true;
                }
            }
            if(removeBrokerName){
                Set<String> nameSet = this.clusterAddrTable.get(clusterName);
                if(null != nameSet){
                    boolean removed = nameSet.remove(brokerName);
                    logger.info("unregisterBroker, remove name from clusterAddrTable {}, {}",
                            removed ? "OK" : "Failed",
                            brokerName);

                    if (nameSet.isEmpty()) {
                        this.clusterAddrTable.remove(clusterName);
                        logger.info("unregisterBroker, remove cluster from clusterAddrTable {}",
                                clusterName
                        );
                    }
                }
            }

        } catch (Exception e) {
            logger.error("unregisterBroker Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }
}
